package com.mxd.flink.connector.operator;

import com.mxd.flink.connector.config.LuaCmd;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.*;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

/**
 * @author rongdi
 * @date 2022/9/19 17:06
 */
public class RedisClusterOperator implements RedisOperator{

    private static final Logger LOG = LoggerFactory.getLogger(RedisClusterOperator.class);

    protected transient JedisCluster jedisCluster;

    private final ExecutorService executorService =
            new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(),
                    0L, TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<>(10000,true));

    public RedisClusterOperator(final JedisCluster jedisCluster) {
        Objects.requireNonNull(jedisCluster, "Jedis cluster can not be null");
        this.jedisCluster = jedisCluster;
    }

    @Override
    public void hset(final String key, final String hashField, final String value) {
        try {
            this.jedisCluster.hset(key, hashField, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command HSET to hash {} of key {} error message {}", hashField, key, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void rpush(final String listName, final String value) {
        try {
            this.jedisCluster.rpush(listName, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command RPUSH to list {} error message: {}", listName, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void rpush(String listName, List<String> values) {
        try {
            this.jedisCluster.rpush(listName, values.toArray(new String[values.size()]));
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command RPUSH to list {} error message: {}", listName, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public List<String> brpop(final String listName) {
        List<String> value;
        try {
            value = this.jedisCluster.brpop(2000, listName);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command RPUSH to list {} error message: {}", listName, e.getMessage());
            }
            throw e;
        }
        return value;
    }

    @Override
    public List<Object> rpop(String listName, long batchFetchRows) {
        List<Object> value = null;
        try {
            String sha = LuaCmd.get(LuaCmd.BATCH_RPOP);
            if(StringUtils.isEmpty(sha)) {
                sha = this.jedisCluster.scriptLoad(LuaCmd.BATCH_RPOP_LUA,LuaCmd.BATCH_RPOP);
            }
            value = (List<Object>)this.jedisCluster.evalsha(sha, Arrays.asList(listName),Arrays.asList(String.valueOf(batchFetchRows)));
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command batch rpop to list {},size {} error message {}", listName,batchFetchRows, e.getMessage());
            }
            throw e;
        }
        return value;
    }

    @Override
    public List<String> blpop(final String listName) {
        List<String> value;
        try {
            value = this.jedisCluster.blpop(2000, listName);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command RPUSH to list {} error message: {}", listName, e.getMessage());
            }
            throw e;
        }
        return value;
    }

    @Override
    public List<Object> lpop(String listName, long batchFetchRows) {
        List<Object> value = null;
        try {
            String sha = LuaCmd.get(LuaCmd.BATCH_LPOP);
            if(StringUtils.isEmpty(sha)) {
                sha = this.jedisCluster.scriptLoad(LuaCmd.BATCH_LPOP_LUA,LuaCmd.BATCH_LPOP);
            }
            value = (List<Object>)this.jedisCluster.evalsha(sha, Arrays.asList(listName),Arrays.asList(String.valueOf(batchFetchRows)));
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command batch rpop to list {},size {} error message {}", listName,batchFetchRows, e.getMessage());
            }
            throw e;
        }
        return value;
    }

    @Override
    public void lpush(final String listName, final String value) {
        try {
            this.jedisCluster.lpush(listName, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command LPUSH to list {} error message: {}", listName, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void lpush(String listName, List<String> values) {
        try {
            this.jedisCluster.lpush(listName, values.toArray(new String[values.size()]));
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command LPUSH to list {} error message: {}", listName, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void sadd(final String setName, final String value) {
        try {
            this.jedisCluster.sadd(setName, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command sadd  {} error message {}", setName, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void sadd(String setName, List<String> values) {
        try {
            this.jedisCluster.sadd(setName, values.toArray(new String[values.size()]));
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command sadd {},size {} error message {}", setName,values.size(), e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public Set<String> spop(String setName, long batchFetchRows) {
        try {
            return this.jedisCluster.spop(setName, batchFetchRows);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command RPUSH to set {} error message {}", setName, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void set(final String key, final String value) {
        try {
            this.jedisCluster.set(key, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command SET to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void pfadd(final String key, final String element) {
        try {
            this.jedisCluster.pfadd(key, element);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command PFADD to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void zadd(final String key, final String score, final String element) {
        try {
            this.jedisCluster.zadd(key, Double.valueOf(score), element);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command ZADD to set {} error message {}", key, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void zrem(final String key, final String element) {
        try {
            this.jedisCluster.zrem(key, element);
        }
        catch (Exception e) {
            if (LOG.isDebugEnabled()) {
                LOG.error("Cannot send Redis message with command ZREM to set {} error message {}", key, e.getMessage());
            }
        }
    }

    @Override
    public void close() throws IOException {
        this.jedisCluster.close();
    }

    @Override
    public void setbit(final String key, final long offset, final boolean value) {
        try {
            this.jedisCluster.setbit(key, offset, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command setbit to key {} with value {} and offset {} error message {}", key, value, offset, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public boolean getbit(final String key, final long offset) {
        try {
            return this.jedisCluster.getbit(key, offset);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command getbit to key {} with offset {} error message {}", key, offset, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public String hget(final String key, final String field) {
        try {
            return this.jedisCluster.hget(key, field);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command hget to key {} with field {} error message {}", key, field, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public CompletableFuture<String> asyncHget(String key, String field) {
        CompletableFuture<String> completableFuture = new CompletableFuture<>();
        executorService.submit(() -> {
            completableFuture.complete(hget(key, field));
        });
        return completableFuture;
    }

    @Override
    public String get(final String key) {
        try {
            return this.jedisCluster.get(key);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command hget to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public CompletableFuture<String> asyncGet(String key) {
        CompletableFuture<String> completableFuture = new CompletableFuture<>();
        executorService.submit(() -> {
            completableFuture.complete(get(key));
        });
        return completableFuture;
    }

    @Override
    public void hdel(final String key, final String field) {
        try {
            this.jedisCluster.hdel(key, field);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command hdel to key {} with field {} error message {}", key, field, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void del(final String key) {
        try {
            this.jedisCluster.del(key);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command del to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public Long expire(final String key, final int seconds) {
        try {
            return this.jedisCluster.expire(key, seconds);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command exists to key {}  seconds {} error message {}", key, seconds, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public boolean sismember(final String key, final String member) {
        try {
            return this.jedisCluster.sismember(key, member);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command exists to key {}  member {} error message {}", key, member, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public long scard(final String key) {
        try {
            return this.jedisCluster.scard(key);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command scard to key {}  error message {}", key, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public void srem(final String setName, final String value) {
        try {
            this.jedisCluster.srem(setName, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command srem to setName {} with value {} error message {}", setName, value, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public long incrByLong(final String key, final long value) {
        long result = 0L;
        try {
            result = this.jedisCluster.incrBy(key, value);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command incrBy to key {} and value {} error message {}", key, value, e.getMessage());
            }
            throw e;
        }
        return result;
    }

    @Override
    public void hset(final String key, final String hashField, final String value, final Integer expireTime) {
        try {
            this.jedisCluster.hset(key, hashField, value);
            if (expireTime != null) {
                this.jedisCluster.expire(key, expireTime);
            }
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command HSET to hash {} of key {} error message {}", hashField, key, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public Map<String, String> hgetAll(final String key) {
        try {
            return this.jedisCluster.hgetAll(key);
        }
        catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error("Cannot send Redis message with command hgetall to key {} error message {}", key, e.getMessage());
            }
            throw e;
        }
    }

    @Override
    public CompletableFuture<Map<String, String>> asyncHgetAll(String key) {
        CompletableFuture<Map<String, String>> completableFuture = new CompletableFuture<>();
        executorService.submit(() -> {
            completableFuture.complete(hgetAll(key));
        });
        return completableFuture;
    }

}
